package com.bitmain.mmst.miner.monitor;

import java.util.ArrayList;
import java.util.Collection;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.TimeZone;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;

import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.collections4.MapUtils;
import org.apache.commons.lang3.ArrayUtils;
import org.apache.commons.lang3.BooleanUtils;

import com.bitmain.mmst.domain.MinerDetectBatch;
import com.bitmain.mmst.domain.MinerDetectDetail;
import com.bitmain.mmst.domain.MinerDetectFailure;
import com.bitmain.mmst.domain.MinerShelf;
import com.bitmain.mmst.domain.MinerState;
import com.bitmain.mmst.domain.MinerType;
import com.bitmain.mmst.feature.cache.GlobalDataBuffer;
import com.bitmain.mmst.feature.time.TimeService;
import com.bitmain.mmst.miner.TaskExecuteEvent;
import com.bitmain.mmst.miner.monitor.executor.MinerCommandExecutor;
import com.bitmain.mmst.miner.monitor.executor.MinerCommandMessage;
import com.bitmain.mmst.miner.monitor.executor.MinerCommandExecutor.Command;
import com.bitmain.mmst.miner.monitor.helper.DetectDataHelper;
import com.bitmain.mmst.miner.monitor.helper.DetectMessageParser;
import com.bitmain.mmst.miner.monitor.internal.MonitorOption;
import com.bitmain.ulnar.aider.DateAider;

/**
 * 
 *
 * @author Richard - rui.wu@bitmain.com
 * @since 2016年7月11日
 *
 */
public class MinerRuntimeDetector extends MinerMonitor{
	
	private Map<String,MinerState> minerStates;
	private Collection<MinerType> types;
	
	private long detectTime;
	private int date;
	
	private final AtomicInteger numAll;
	private final AtomicInteger numDetect;
	private final AtomicInteger numDetail;
	private final AtomicInteger numFailure;
	
	MinerRuntimeDetector(){
		this.minerStates = MinerStateCache.INSTANCE.getPoisedMinerCache();
		this.numAll = new AtomicInteger(0);
		this.numDetect = new AtomicInteger(0);
		this.numDetail = new AtomicInteger(0);
		this.numFailure = new AtomicInteger(0);
	}
	
	void execute(){
		detectTime = TimeService.getTimeMillis();
		TimeZone timeZone = TimeService.getTimeZone();
		String datePattern = "yyyyMMdd";
		
		String detectDate = DateAider.formatDate(new Date(detectTime), timeZone, datePattern);
		date = Integer.parseInt(detectDate);
		
		long dateBeginning = DateAider.parseMillis(detectDate, timeZone, datePattern);
		long dateEnd = dateBeginning + 24*60*60*1000;
		
		if(DetectDataHelper.detailTableExists(date)){
			System.out.println("数据表存在");
		}else{
			System.out.println("数据表不存在，即将创建数据表,date,dateBeginning,dateEnd:"+date+dateBeginning+dateEnd);
			DetectDataHelper.createDetailTable(date,dateBeginning,dateEnd);
		}
		
		if(MapUtils.isEmpty(minerStates)){
			System.out.println("【运行时状态检测】1、矿机状态缓存未准备就绪，退出待下一周执行");
			return;
		}
		System.out.println("【运行时状态检测】1、从缓存中获取矿机的个数为：" + minerStates.size());

		List<MinerShelf> shelfs = DetectDataHelper.fetchAllShelfs();
		List<MinerCommandMessage> messages = new ArrayList<MinerCommandMessage>();
		Map<String,MinerState> minerStates = MinerStateCache.INSTANCE.getPoisedMinerCache();
		types = GlobalDataBuffer.getMinerTypes();
		
		for(MinerShelf shelf:shelfs){
			int numIgnore = 0;
			Integer[] ignores = shelf.getIgnores();
			if(ArrayUtils.isNotEmpty(ignores)){
				numIgnore = ignores.length;
			}
			int IpBeginning = shelf.getIpBeginning();
			int IpEnd = shelf.getIpEnd();
			numAll.addAndGet(IpEnd - IpBeginning + 1 - numIgnore);
			
			for(int i=IpBeginning;i<=IpEnd;i++){
				String ip = shelf.getIpBlock()+"."+i;
				if(numIgnore>0 && ArrayUtils.contains(ignores, i)){
					continue;
				}
				MinerState minerState = minerStates.get(ip);
				
				if(minerState==null){
					continue;
				}
				if(BooleanUtils.isNotTrue(minerState.getMacable())){
					numFailure.incrementAndGet();
					MinerDetectFailure failure = new MinerDetectFailure();
					failure.setMinerShelfId(shelf.getId());
					failure.setMinerHost(minerState.getHost());
					failure.setMinerMac(minerState.getMac());
					failure.setPingable(minerState.getPingable());
					failure.setMacable(minerState.getMacable());
					failure.setDetectTime(detectTime);
					DetectDataHelper.saveFailure(failure);
					continue;
				}
				MinerCommandMessage message = new MinerCommandMessage();
				message.setShelfId(shelf.getId());
				message.setMinerTypeId(shelf.getMinerTypeId());
				message.setHost(ip);
				message.setPort(shelf.getMinerTypePort());
				
				String mac = minerState.getMac();
				message.setHost(minerState.getHost());
				message.setMac(mac);
				message.setPingable(true);
				message.setMacable(true);
				message.setExecuteTime(detectTime);
				messages.add(message);
			}
		}
		allocateExecutor(messages);
	}
	
	private void allocateExecutor(List<MinerCommandMessage> messages) {
		if(CollectionUtils.isEmpty(messages)){
			System.out.println("【运行时状态检测】所有矿机都ping都不能连通");
			return;
		}
		int numThread = MonitorOption.runtimeThreadSize;
		if(messages.size() <= numThread){
			executeOverall(messages,numThread);
		}else{
			executeBatching(messages,numThread);
		}
	}
	
	private void executeOverall(List<MinerCommandMessage> messages, int numThread) {
		ExecutorService service = Executors.newFixedThreadPool(messages.size());
		System.out.println("【运行时状态检测】2、矿机总数："+ messages.size());
		for(MinerCommandMessage message:messages){
			submitExecutor(message,service);
		}
		service.shutdown();
		boolean done = false;
		while(!done){
			if(service.isTerminated()){
				done = true;
				service.shutdownNow();
				System.out.println("service线程池执行结束");
				executeDone();
			}else{
				try {
					Thread.sleep(1000);
				} catch (Exception e) {
				}
			}
		}
	}
	
	private void executeBatching(List<MinerCommandMessage> messages, int numThread) {
	int numBatch = (messages.size() + numThread -1) / numThread;
		
		System.out.println("【运行时状态检测】2、矿机总数为："+ messages.size()+"，最大线程数为："+numThread+"，执行executeBatching方法,分批数为："+numBatch);
		for(int i=0;i<numBatch;i++){
			int fromIndex = i*numThread;//0,100,200
			int toIndex = fromIndex + numThread;//99,199
			System.out.println("【运行时状态检测】第" + i +"批，从"+fromIndex+"--"+toIndex);
			if(toIndex >= messages.size()){
				toIndex = messages.size();
			}
			List<MinerCommandMessage> batchMessages = messages.subList(fromIndex, toIndex);
			ExecutorService service = Executors.newFixedThreadPool(numThread);
			for(MinerCommandMessage message:batchMessages){
				submitExecutor(message , service);
			}
			service.shutdown();
			long start = System.currentTimeMillis();
			boolean done = false;
			while (!done) {
				if (service.isTerminated()) {
					done = true;
					service.shutdownNow();
					executeDone();
					System.out.println("\r\n\tNetworkDetector : batch (" + (fromIndex + 1) + "-" + toIndex + ") - complete.");
				} else {
					if (System.currentTimeMillis() - start > 1000 * 60) {
						done = true;
						service.shutdownNow();
						System.out.println("\r\n\tNetworkDetector ：batch (" + (fromIndex + 1) + "-" + toIndex + ") - timeout.");
					}
				}
				try {
					Thread.sleep(1000);
				} catch (InterruptedException e) {
				}
			}
		}
	}
	
	private void executeDone() {
		long elapsedTime = TimeService.getTimeMillis() - detectTime;
		
		MinerDetectBatch batch = new MinerDetectBatch();
		batch.setNumAll(numAll.get());
		batch.setNumDetect(numDetect.get());
		batch.setNumDetail(numDetail.get());
		batch.setNumFailure(numFailure.get());
		batch.setDetectTime(detectTime);
		batch.setElapsedTime(elapsedTime);
		batch.setSynced(false);
		
		System.out.println("运行时状态报告：");
		System.out.println("矿机总数:" + batch.getNumAll());
		System.out.println("检测到矿机:" + batch.getNumDetect());
		System.out.println("检测成功数：" + batch.getNumDetail());
		System.out.println("检测失败数：" + batch.getNumFailure());
		System.out.println("是否提交到minerlink中心：" + batch.getSynced());
		DetectDataHelper.saveBatch(batch);
		TaskExecuteEvent event = new TaskExecuteEvent();
		event.data = batch;
		executeDone(event);
	}

	private void submitExecutor(MinerCommandMessage message, ExecutorService service) {
		MinerCommandExecutor executor = new MinerCommandExecutor(message,Command.ALL.key);
		CompletableFuture<MinerCommandMessage> future = CompletableFuture.supplyAsync(executor, service);
		future.thenAccept(m -> processMessage(m));
	}
	private void processMessage(MinerCommandMessage message) {
		numDetect.getAndIncrement();
		
		MinerState minerState = MinerStateCache.INSTANCE.pull(message.getHost());
		if (minerState != null) {
			minerState.setStatsable(message.getStatsable());
			minerState.setPoolsable(message.getPoolsable());
			minerState.setSummaryable(message.getSummaryable());
//			System.out.println(message.getHost()+"三种状态设置完毕"+minerState.getStatsable()+minerState.getPoolsable()+minerState.getSummaryable());
		}
		
		if(BooleanUtils.isTrue(message.getCommandable())){
			numDetail.incrementAndGet();
			MinerDetectDetail detail = DetectMessageParser.processMessage(message, types);
			
			if(detail!=null){
				if (minerState != null) {
					minerState.setPools(detail.getPools());
					minerState.setFirmwareVersion(detail.getFirmwareVersion());
					minerState.setMinerType(detail.getMinerType());
//					System.out.println("矿池、版本信息、矿机型号信息设置完毕"+minerState.getFirmwareVersion()+minerState.getMinerType());
				}
				detail.setDate(date);
				DetectDataHelper.saveDetail(detail);
				System.out.println(detail.getMinerHost()+"detail信息存储完成");
			}
		}else{
			numFailure.incrementAndGet();
			MinerDetectFailure failure = DetectMessageParser.processFailure(message);
			
			DetectDataHelper.saveFailure(failure);
			System.out.println(failure.getMinerHost()+"failure信息存储完成");
		}
	}
}
